package demo.designPattern.observer;

import demo.java.lang.ThreadDemo;
import demo.java.util.concurrent.ThreadFactoryDemo;
import org.junit.Test;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Reactive Streams 的设计目标是支持异步数据流处理，适用于高并发、低延迟的场景。
 */
public class ReactiveStreamDemo {

    static Logger logger = LoggerFactory.getLogger(ReactiveStreamDemo.class);

    static ThreadPoolExecutor producerExecutor = new ThreadPoolExecutor(3, 3, 0, TimeUnit.MINUTES,
            new ArrayBlockingQueue<>(10),
            new ThreadFactoryDemo.NamedThreadFactory("producer", true),
            new ThreadPoolExecutor.AbortPolicy());

    static ThreadPoolExecutor consumerExecutor = new ThreadPoolExecutor(3, 3, 0, TimeUnit.MINUTES,
            new ArrayBlockingQueue<>(10),
            new ThreadFactoryDemo.NamedThreadFactory("consumer", true),
            new ThreadPoolExecutor.AbortPolicy());
    /**
     * Subscriber 数据消费者，负责接收和处理数据。它有以下方法：
     *
     * <li>onSubscribe(Subscription s)：在订阅时调用，接收一个 Subscription 对象。
     * <li>onNext(T item)：接收到新数据时调用。
     * <li>onError(Throwable t)：发生错误时调用。
     * <li>onComplete()：数据流结束时调用。
     */
    static Subscriber<Integer> subscriber = new Subscriber<Integer>() {
        /**
         * Subscription 表示消费者与生产者之间的订阅关系。它有以下方法：
         * <li>request(long n)：消费者请求一定数量的数据。
         * <li>cancel()：消费者取消订阅。
         */
        private Subscription subscription;

        @Override
        public void onSubscribe(Subscription s) {
            this.subscription = s;
            logger.info("订阅成功");
            s.request(1); // 请求一个数据
        }

        @Override
        public void onNext(Integer item) {
            consumerExecutor.execute(() -> {
                logger.info("收到数据：{}", item);
                subscription.request(1); // 继续请求下一个数据
            });
        }

        @Override
        public void onError(Throwable t) {
            t.printStackTrace();
        }

        @Override
        public void onComplete() {
            logger.info("数据流结束");
        }
    };
    /**
     * Publisher：数据生产者，负责生成数据流。它有一个 subscribe 方法，允许消费者订阅数据流。
     */
    static Publisher<Integer> publisher = new Publisher<Integer>() {
        @Override
        public void subscribe(Subscriber<? super Integer> subscriber) {
            subscriber.onSubscribe(new Subscription() {
                private int count = 0;

                @Override
                public void request(long n) {
                    producerExecutor.execute(() -> {
                        logger.info("请求数量：{}", n);
                        for (int i = 0; i < n; i++) {
                            if (count < 10) {
                                subscriber.onNext(count++);
                            } else {
                                subscriber.onComplete();
                                break;
                            }
                        }
                    });
                }

                @Override
                public void cancel() {
                    logger.info("取消订阅");
                    // 取消订阅
                }
            });
        }
    };

    @Test
    public void test1() {
        publisher.subscribe(subscriber);
        ThreadDemo.safeSleep(2000);
    }
}
